package com.pai4j.connect.messagequeue.consumer.service;
import com.pai4j.common.constants.ConnectConstants;
import com.pai4j.common.enums.RoleCodeEnum;
import com.pai4j.common.enums.messagequeue.MessageQueueEnum;
import com.pai4j.common.service.messagequeue.producer.MessageQueueProducer;
import com.pai4j.common.util.JsonUtil;
import com.pai4j.connect.service.AccountService;
import com.pai4j.connect.websocket.handler.ChatWebSocketEndPointServletHandler;
import com.pai4j.domain.vo.request.mess.ChatMessageRequestVO;
import com.pai4j.domain.vo.response.UserBaseResponseInfoVO;
import com.pai4j.domain.vo.response.mess.ChatMessageResponseVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.Date;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;


/**
 * 负责处理对话WS请求消息，来自广播
 * （不包含音视频、红包等，文字对话+红包+文件传输等）
 */
@Slf4j
@Service
public class ChatMessageConsumeService {

    @Autowired
    private AccountService accountService;

    @Autowired
    private MessageQueueProducer<ChatMessageRequestVO, Long> messageQueueProducer;
    

    public void handleChatMessage(String message) {
        this.handleChatMessage(null, message);
    }

    /**
     * 处理来自对话topic 广播的消息
     * @param message
     */
    public void handleChatMessage(String bizCode, String message) {
        log.info("处理对话消息 =====> {}", message);
        ChatMessageRequestVO chatMessage = JsonUtil.fromJson(message, ChatMessageRequestVO.class);
        log.info("反序列化后的消息 =====> {}", chatMessage);
        if (chatMessage == null) {
            return;
        }
        // 获取对话消息接受人
        String receiverId = chatMessage.getReceiverId();
        UserBaseResponseInfoVO receiver = accountService.accountInfo(receiverId);
        if (receiver == null) {
            return;
        }
        // 最终要推送的用户
        Set<String> receivers = new HashSet<>();
        /**
         * 处理一对一、群聊
         */
        // 构建消息响应VO，用于推送客户端
        ChatMessageResponseVO messageResponse = new ChatMessageResponseVO();
        messageResponse.setMessage(chatMessage.getMessage());
        messageResponse.setSenderId(chatMessage.getSenderId());
        messageResponse.setReceiverId(chatMessage.getReceiverId());
        messageResponse.setTime(new Date().getTime());
        messageResponse.setReceiverId(receiverId);
        messageResponse.setReceiver(receiver);
        messageResponse.setType(chatMessage.getType());
        messageResponse.setImage(chatMessage.getImage());
        if (RoleCodeEnum.GROUP.getCode().equals(receiver.getRoleCode())) {
            /**
             * 群聊(消息代理人，角色反转，原消息接受人-群聊反转为消息发送人，实际消息发送人作为被代理发送人)
             */
            // 由群组代理消息发送
            messageResponse.setSenderId(receiverId);
            // 被代理消息发送人
            messageResponse.setProxySenderId(chatMessage.getSenderId());
            UserBaseResponseInfoVO sender = accountService.accountInfo(chatMessage.getSenderId());
            messageResponse.setProxySender(sender);
            Set<String> groupUserIds = accountService.getGroupUserIds(receiverId);
            receivers.addAll(groupUserIds);
        } else if (RoleCodeEnum.isUserRole(receiver.getRoleCode())) {
            /**
             * 用户账号间一对一对话
             */
            receivers.add(receiverId);
        } else {
            log.error("不支持的对话场景 =====> receiverId:{}", receiverId);
            return;
        }

        /**
         * 《实时场景》消息推送
         * 
         * 【关键修复】只有成功推送消息到至少一个在线用户的节点才发送到持久化队列
         * 这样可以避免分布式环境下多个节点重复发送持久化请求
         */
        boolean messagePushed = false;
        if (StringUtils.isNotBlank(bizCode)) {
            messagePushed = this.doPushMessage2Client(bizCode, messageResponse, receivers);
        } else {
            messagePushed = this.doPushMessage2Client(messageResponse, receivers);
        }
        
        /**
         * 《离线场景》持久化消息队列，保存离线消息，同时会将数据同步到ES用于后续的RAG问答和消息搜索
         *
         * 【重要修复】只有以下情况才发送到持久化队列：
         * 1. 消息成功推送到至少一个在线用户（messagePushed=true）
         * 2. 所有接收者都离线（messagePushed=false），此时需要持久化以便用户上线后查看
         * 
         * 这样可以确保在分布式环境下：
         * - 如果有用户在线，只有成功推送的节点会发送持久化请求（避免重复）
         * - 如果所有用户都离线，所有节点都会发送持久化请求，但UGC服务的幂等性检查会防止重复入库
         * 
         * Consumer在UGC服务：
         * com.pai4j.ugc.messaegqueue.consumer.redis.PersistentMessageRedisMQConsumer
         * com.pai4j.ugc.messaegqueue.consumer.rocketmq.PersistentMessageRocketMQConsumer
         */
        if (messagePushed || !hasAnyOnlineReceiver(receivers)) {
            messageQueueProducer.send(MessageQueueEnum.QUEUE_PERSISTENT_CHAT_MESSAGE, chatMessage);
            log.debug("消息已发送到持久化队列。messagePushed: {}, receivers: {}", messagePushed, receivers.size());
        } else {
            log.debug("消息未发送到持久化队列（其他节点已处理）。receivers: {}", receivers.size());
        }
    }


    /**
     * 推送消息到指定bizCode的客户端
     * 
     * @param bizCode 业务代码
     * @param chatMessage 聊天消息
     * @param receivers 接收者集合
     * @return 是否成功推送到至少一个在线用户
     */
    private boolean doPushMessage2Client(String bizCode, ChatMessageResponseVO chatMessage, Set<String> receivers) {
        Map<String, WebSocketSession> wsSessionMap = ChatWebSocketEndPointServletHandler.getSessions(bizCode, receivers);
        if (MapUtils.isEmpty(wsSessionMap)) {
            return false;
        }
        
        boolean anyMessageSent = false;
        chatMessage.setTime(System.currentTimeMillis());
        for (String receiver : receivers) {
            if (receiver.equals(chatMessage.getProxySenderId())) {
                continue;
            }
            chatMessage.setReceiverId(receiver);
            String message = JsonUtil.toJsonString(chatMessage);
            WebSocketSession socketSession = wsSessionMap.get(receiver);
            if (socketSession == null || !socketSession.isOpen()) {
                // 当前接受人可能离线、或者当前接受人的ws 链接在其他节点（
                // 广播后由其他节点处理 - 分布式下webscoket session共享，相见视频讲解）
                continue;
            }
            try {
                socketSession.sendMessage(new TextMessage(message));
                anyMessageSent = true; // 标记至少成功推送了一条消息
                log.debug("消息成功推送到用户 {}", receiver);
            } catch (IOException e) {
                log.error("聊天消息推送异常 =====> receiver:{}, message:{}", receiver, message, e);
                // 这里如果异常可以增加重试策略 TODO
                // 但是不能阻塞其他人的消息推送（在多人群聊场景）
            }
        }
        return anyMessageSent;
    }

    /**
     * 推送消息到所有bizCode的客户端
     *
     * @param chatMessage 聊天消息
     * @param receivers 接收者集合
     * @return 是否成功推送到至少一个在线用户
     */
    private boolean doPushMessage2Client(ChatMessageResponseVO chatMessage, Set<String> receivers) {
        Set<String> bizCodes = ConnectConstants.ConnectBiz.getBizCode(ConnectConstants.BizEnum.CHAT);
        boolean anyMessageSent = false;
        for (String bizCode : bizCodes) {
            boolean sent = this.doPushMessage2Client(bizCode, chatMessage, receivers);
            anyMessageSent = anyMessageSent || sent;
        }
        return anyMessageSent;
    }
    
    /**
     * 检查接收者中是否有在线用户
     * 
     * @param receivers 接收者集合
     * @return 是否有在线用户
     */
    private boolean hasAnyOnlineReceiver(Set<String> receivers) {
        Set<String> bizCodes = ConnectConstants.ConnectBiz.getBizCode(ConnectConstants.BizEnum.CHAT);
        for (String bizCode : bizCodes) {
            Map<String, WebSocketSession> wsSessionMap = ChatWebSocketEndPointServletHandler.getSessions(bizCode, receivers);
            if (MapUtils.isNotEmpty(wsSessionMap)) {
                for (WebSocketSession session : wsSessionMap.values()) {
                    if (session != null && session.isOpen()) {
                        return true;
                    }
                }
            }
        }
        return false;
    }

}
